package march.finalmarch21.march01.plan2;

import march.finalmarch21.FlowBean;
import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import java.io.BufferedReader;
import java.io.FileInputStream;
import java.io.IOException;
import java.io.InputStreamReader;
import java.math.BigDecimal;
import java.net.URI;
import java.util.HashMap;

public class MarchMapper extends Mapper<LongWritable, Text, FlowBean, NullWritable> {
    HashMap<String, BigDecimal> huilvMap = new HashMap();
    HashMap<String, String> huilvNameMap= new HashMap();



    @Override
    protected void setup(Context context) throws IOException, InterruptedException {
        URI[] cacheFiles = context.getCacheFiles();
        String line;

        String path1 = cacheFiles[0].getPath();
        BufferedReader reader = new BufferedReader(new InputStreamReader(new FileInputStream(path1), "UTF-8"));
        while(StringUtils.isNotEmpty(line = reader.readLine())){
            String[] split = line.split(",");
            huilvMap.put(split[0],new BigDecimal(split[1]));
            huilvNameMap.put(split[0],split[2]);

        }

        IOUtils.closeStream(reader);
    }

    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        FlowBean flowBean = new FlowBean();
        String[] split = value.toString().split(",");

        flowBean.setId(split[0]);
        flowBean.setIndustry(split[1]);
        flowBean.setType(split[2]);
        flowBean.setCoinId(split[3]);
        flowBean.setLoanAmount(split[4]);
        flowBean.setDrawMoneyAmount(split[5]);
        flowBean.setRepaidAmount(split[6]);
        flowBean.setInterestAmount(split[7]);

        BigDecimal huiLv = huilvMap.get(flowBean.getCoinId());
        String coinName = huilvNameMap.get(flowBean.getCoinId());

        //考虑到汇率计算，所以在这里使用BigDecimal，而不是使用浮点数型
            BigDecimal loanRMB = new BigDecimal(flowBean.getLoanAmount()).multiply(huiLv);
            BigDecimal drawRMB = new BigDecimal(flowBean.getDrawMoneyAmount()).multiply(huiLv);
            BigDecimal repaidRMB = new BigDecimal(flowBean.getRepaidAmount()).multiply(huiLv);
            BigDecimal interestRMB = new BigDecimal(flowBean.getInterestAmount()).multiply(huiLv);
        // 贷款余额=实际提款额-已还本金额
            BigDecimal balanceRMB = drawRMB.subtract(repaidRMB);

            //数据填充
        flowBean.setExchangePrice(huiLv.toString());
        flowBean.setRepaidRMB(repaidRMB.toString());
        flowBean.setCoinName(coinName);
        flowBean.setLoanRMB(loanRMB.toString());
        flowBean.setDrawRMB(drawRMB.toString());
        flowBean.setInterestRMB(interestRMB.toString());
        flowBean.setBalanceRMB(balanceRMB.toString());

        context.write(flowBean,NullWritable.get());
    }
}
